﻿#include "include/gtest.h"
#include "../Tool/include/BlockQueue.h"
#include "../Tool/include/ILogger.h"
#include "../Tool/include/Timer.h"
#include <iostream>
#include <thread>

using namespace Tool;

std::atomic<uint64_t> g_total_processed(0);
void Producer(BlockQueue<uint64_t>& queue, int numItems)
{
    for (int i = 1; i <= numItems; ++i)
    {
        queue.Push(i);
    }
}

void Consumer(BlockQueue<uint64_t>& queue, int numItems)
{
    while (queue.Running())
    {
        if(queue.Pop())
        {
            g_total_processed++;
        }
    }
}

TEST(BlockQueue, ThroughputTest_500k) 
{
    Timer timer;
    const size_t QUEUE_MAX_SIZE = 1000;
    const uint64_t TOTAL_ITEMS = 5000000;    // 总任务数：500万
    const size_t PRODUCER_NUM = 4;
    const size_t CONSUMER_NUM = 4;
    const uint64_t ITEMS_PER_PRODUCER = TOTAL_ITEMS / PRODUCER_NUM;
    const uint64_t ITEMS_PER_CONSUMER = TOTAL_ITEMS / CONSUMER_NUM;

    BlockQueue<uint64_t> queue(QUEUE_MAX_SIZE);
    std::vector<std::thread> producers;
    std::vector<std::thread> consumers;

    g_total_processed.store(0, std::memory_order_relaxed);
    timer.Reset();

    for (size_t i = 0; i < PRODUCER_NUM; ++i) 
    {
        producers.emplace_back(Producer, std::ref(queue), ITEMS_PER_PRODUCER);
    }

    for (size_t i = 0; i < CONSUMER_NUM; ++i) 
    {
        consumers.emplace_back(Consumer, std::ref(queue), ITEMS_PER_CONSUMER);
    }

    for (auto& t : producers) 
    {
        t.join();
    }
    queue.Stop();
    for (auto& t : consumers) 
    {
        t.join();
    }

    double elapsed_ms = timer.Elapsed();
    double throughput = TOTAL_ITEMS / (elapsed_ms / 1000);

    if(g_total_processed.load() != TOTAL_ITEMS)
    {
        LOG_ERROR("队列丢数据！实际处理：%llu", g_total_processed.load());
    }

    LOG_INFO("=== BlockQueue %d万吞吐测试结果 ===", TOTAL_ITEMS / 10000);
    LOG_INFO("总任务数：%llu", TOTAL_ITEMS);
    LOG_INFO("总耗时：%.3f ms", elapsed_ms);
    LOG_INFO("吞吐量：%.0f 次/秒", throughput);
    LOG_INFO("======================================");
}
